package com.im.gateway.tcp.dispatcher;

import com.im.common.ImConstant;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 * @Created leijun
 * @Date 2021/9/12 9:10 下午
 */
public class DispatcherInstanceManager {
    // 分发系统服务集合，暂时写死，可以从zk中获取
    private static final List<DispatcherAddress> dispatcherAddresses = new ArrayList<>();
    /**
     * 保存分发系统的连接
     */
    private Map<String, DispatcherInstance> dispatcherInstances = new ConcurrentHashMap<>();

    static {
        dispatcherAddresses.add(new DispatcherAddress("127.0.0.1", "localhost", 8090));
    }

    static class Singleton {
        private static DispatcherInstanceManager instance = new DispatcherInstanceManager();
    }

    public static DispatcherInstanceManager getInstance() {
        return Singleton.instance;
    }

    /**
     * 初始化
     */
    public void init() {
        // 与每一个分发系统都建立连接
        for (DispatcherAddress dispatcherAddress : dispatcherAddresses) {
            System.out.println("与分发系统"+dispatcherAddress+"建立连接");
            connect(dispatcherAddress);
        }
    }

    /**
     * 与分发系统建立连接
     * @param dispatcherAddress
     */
    private void connect(DispatcherAddress dispatcherAddress) {
        EventLoopGroup eventLoopGroup = new NioEventLoopGroup(1);
        try {
            Bootstrap dispatcherBootstrap = new Bootstrap();
            dispatcherBootstrap.group(eventLoopGroup)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .option(ChannelOption.SO_KEEPALIVE, true)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ByteBuf delimiter = Unpooled.copiedBuffer(ImConstant.DELIMITER);
                            socketChannel.pipeline().addLast(new DelimiterBasedFrameDecoder(4096, delimiter));
                            socketChannel.pipeline().addLast(new DispatcherInstanceHandler());
                        }
                    });
            ChannelFuture channelFuture = dispatcherBootstrap.connect(dispatcherAddress.getHost(), dispatcherAddress.getPort()).sync();
            channelFuture.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    if (channelFuture.isSuccess()) {
                        SocketChannel socketChannel = (SocketChannel) channelFuture.channel();
                        String dispatcherChannelId = socketChannel.remoteAddress().getHostName()
                                + ":"
                                + socketChannel.remoteAddress().getPort();
                        dispatcherInstances.put(dispatcherChannelId, new DispatcherInstance(socketChannel, dispatcherAddress));
                    }
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 随机选择一个分发系统实例
     * @return
     */
    public DispatcherInstance chooseDispatcherInstance() {
        Random random = new Random(dispatcherInstances.size());
        int index = random.nextInt(dispatcherInstances.size());
        List<DispatcherInstance> instances = new ArrayList<>(dispatcherInstances.values());
        return instances.get(index);
    }

    /**
     * 移除分发系统实例
     * @param dispatcherChannelId
     */
    public void remove(String dispatcherChannelId) {
        dispatcherInstances.remove(dispatcherChannelId);
    }
}
